package com.lesent.kafka.spring.listener;

import com.lesent.kafka.spring.Constant.Constant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author shengtao
 * @Description:
 * @date 2018/12/20 14:29
 */
@Component
public class ConsumerListener {

//    @KafkaListener(id = "default-listener-26",containerFactory = "kafkaListenerContainerFactory",
//                    topics = Constant.DEFAULT_TOPIC,containerGroup = "group_1",errorHandler = "kafkaErrorHandler")
    public void listenDefaultTopic(ConsumerRecord<String, String> record){
        System.out.printf("------------offset = %d, key = %s , patition = %d, value = %s", record.offset(), record.serializedKeySize(),record.partition(), record.serializedValueSize());
        System.out.println("--");
        Header[] headers = record.headers().toArray();
        for (int i=0;i<headers.length;i++){
            System.out.println("key ==>"+headers[i].key()+"  value ==>"+headers[i].value());
        }
    }
//    @KafkaListener(id = "default-listener-23",containerFactory = "kafkaListenerContainerFactory",
//            topics = Constant.DEFAULT_TOPIC,containerGroup = "group_3",errorHandler = "kafkaErrorHandler")
//    public void listen(@Payload String value,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partition,
//                       @Header(KafkaHeaders.OFFSET)long offset,
////                       @Header(KafkaHeaders.RAW_DATA)String data,
//                       @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) int key,
//                       @Header(KafkaHeaders.CONSUMER)KafkaConsumer<String,String> consumer,
//                       @Header(KafkaHeaders.RECEIVED_TOPIC)List<String> topic,
//                       @Header(KafkaHeaders.RECEIVED_TIMESTAMP)List<String> timeStamp){
////        String patitio = partition.stream().collect(joining(","));
//        System.out.printf("------------offset = %d, topic = %s , timeStamp = %s, value = %s ,key=%d", offset, topic,timeStamp, value,key);
//    }

}
